package com.flink.examples.kafka;

import com.flink.examples.TUser;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * @Description 从Kafka中消费数据
 * @Author JL
 * @Date 2020/09/18
 * @Version V1.0
 */
public class DataStreamSource {

    /**
     * 官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/kafka.html
     */

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度（使用几个CPU核心）
        env.setParallelism(1);
        //每隔2000ms进行启动一个检查点
        env.enableCheckpointing(2000);
        //设置模式为exactly-once
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 确保检查点之间有进行500 ms的进度
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        //1.消费者客户端连接到kafka
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.35:9092");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 5000);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-45");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), props);
        //setStartFromEarliest()会从最早的数据开始进行消费，忽略存储的offset信息
        //consumer.setStartFromEarliest();
        //Flink从topic中指定的时间点开始消费，指定时间点之前的数据忽略
        //consumer.setStartFromTimestamp(1559801580000L);
        //Flink从topic中最新的数据开始消费
        //consumer.setStartFromLatest();
        //Flink从topic中指定的group上次消费的位置开始消费，所以必须配置group.id参数
        //consumer.setStartFromGroupOffsets();

        //2.在算子中进行处理
        DataStream<TUser> sourceStream = env.addSource(consumer)
                .filter((FilterFunction<String>) value -> StringUtils.isNotBlank(value))
                .map((MapFunction<String, TUser>) value -> {
                    System.out.println("print:" + value);
                    //注意，因已开启enableCheckpointing容错定期检查状态机制，当算子出现错误时,
                    //会导致数据流恢复到最新checkpoint的状态，并从存储在checkpoint中的offset开始重新消费Kafka中的消息。
                    //因此会有可能导制数据重复消费，重复错误，陷入死循环。加上try|catch，捕获错误后再正确输出。
                    try {
                        TUser user = new Gson().fromJson(value, TUser.class);
                        return user;
                    } catch (Exception e) {
                        System.out.println("error：" + e.getMessage());
                    }
                    return new TUser();
                })
                .returns(TUser.class);
        sourceStream.print();

        //3.执行
        env.execute("flink  kafka source");
    }

}
